package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/3/5
 */
public class Demo13_TVFWindowCumulate
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs());

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            //在流中产生水印
            .assignTimestampsAndWatermarks(watermarkStrategy);

        Table table = tableEnv.fromDataStream(ds, $("id"),$("ts"),$("vc"),
            $("pt").proctime(),
            $("et").rowtime());

        /*
            累积窗口适用于以下场景:
                    没间隔1h，统计今天的所有的PV。
                    窗口: [0,1], [0,2].... [0,24], [第二天,0,1]

                CUMULATE(TABLE data, DESCRIPTOR(timecol), step, size)
                    slide:  1h
                    max size: 24h
         */

        tableEnv.createTemporaryView("t1",table);

        // [0,2),[0,4),[0,6) | [6,8),[6,10),[6,12) | [12,14).....
        String sql1 = " select  window_start, window_end, id,sum(vc) sumVc " +
            "  from TABLE(" +
            "    CUMULATE(TABLE t1, DESCRIPTOR(et), INTERVAL '2' SECONDS, INTERVAL '6' SECONDS)) " +
            "  GROUP BY window_start, window_end,id ";

        tableEnv.sqlQuery(sql1)
                        .execute()
                        .print();

    }
}
